NexusPi Git Node
Commit 4d3fa4cf8755d2b591f6adfaad1577626da43cf1
Parents : e834170
Author : Ivan <e318cbc04468bd574db2b4523dddd710>
Signature : T66BB85Valid, signed by author
Date : 2026-08-14T12:13:10-05:00
feat: update community interfaces JSON builder with URL validation and fetching logic
Changes
14 files changed, 273 insertions(+), 740 deletions(-)
Diff
diff --git a/meshchatx.rsm b/meshchatx.rsm
index b691e225..f8632ba3 100644
Binary files a/meshchatx.rsm and b/meshchatx.rsm differ
diff --git a/scripts/build_community_interfaces_json.py b/scripts/build_community_interfaces_json.py
index 13362174..b4f826df 100644
--- a/scripts/build_community_interfaces_json.py
+++ b/scripts/build_community_interfaces_json.py
@@ -7,11 +7,11 @@ import argparse
import json
import sys
import urllib.error
+import urllib.request
from pathlib import Path
+from urllib.parse import urlparse
from meshchatx.src.backend.community_interfaces_directory import (
- DEFAULT_SUBMITTED_URL,
- build_interfaces_from_directory_url,
rows_from_payload,
transform_directory_rows,
)
@@ -19,6 +19,98 @@ from meshchatx.src.backend.community_interfaces_directory import (
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "meshchatx" / "src" / "backend" / "data" / "community_interfaces.json"
+DEFAULT_SUBMITTED_URL = (
+ "https://directory.rns.recipes/api/directory/submitted?status=online"
+)
+DEFAULT_DISCOVERED_URL = (
+ "https://directory.rns.recipes/api/directory/discovered?status=online"
+)
+DEFAULT_DIRECTORY_URLS = (DEFAULT_SUBMITTED_URL, DEFAULT_DISCOVERED_URL)
+_ALLOWED_HOST = "directory.rns.recipes"
+_MAX_FETCH_BYTES = 1 * 1024 * 1024
+_FETCH_HEADERS = {
+ "Accept": "application/json",
+ "User-Agent": "MeshChatX-community-interfaces-build/1.0 (+https://meshchatx.com/)",
+}
+
+
+def validate_directory_fetch_url(url: str) -> str:
+ if not url or not isinstance(url, str):
+ msg = "URL must be a non-empty string"
+ raise ValueError(msg)
+ parsed = urlparse(url.strip())
+ if parsed.scheme != "https":
+ msg = "Community directory URL must use https"
+ raise ValueError(msg)
+ netloc = parsed.netloc or ""
+ if "@" in netloc:
+ msg = "Community directory URL must not contain credentials"
+ raise ValueError(msg)
+ host = (parsed.hostname or "").lower()
+ if host != _ALLOWED_HOST:
+ msg = "Community directory URL host is not allowed"
+ raise ValueError(msg)
+ return url.strip()
+
+
+def fetch_directory_payload(url: str, *, timeout: float = 60.0) -> object:
+ resolved = validate_directory_fetch_url(url)
+ req = urllib.request.Request(resolved, headers=_FETCH_HEADERS, method="GET")
+ with urllib.request.urlopen(req, timeout=timeout) as resp:
+ raw = resp.read(_MAX_FETCH_BYTES + 1)
+ if len(raw) > _MAX_FETCH_BYTES:
+ msg = f"Community directory download exceeds {_MAX_FETCH_BYTES} bytes"
+ raise ValueError(msg)
+ return json.loads(raw.decode("utf-8"))
+
+
+def _merge_directory_rows(row_lists: list[list]) -> list:
+ merged: list = []
+ seen: set[tuple] = set()
+ for rows in row_lists:
+ for row in rows:
+ if not isinstance(row, dict):
+ continue
+ key = (
+ str(row.get("name") or "").strip().lower(),
+ str(row.get("type") or "").strip().lower(),
+ str(row.get("host") or row.get("address") or "").strip().lower(),
+ str(row.get("port") or "").strip(),
+ )
+ if key in seen:
+ continue
+ seen.add(key)
+ merged.append(row)
+ return merged
+
+
+def build_interfaces_from_directory_url(
+ url: str | None = None,
+ *,
+ timeout: float = 60.0,
+) -> tuple[list[dict], str]:
+ if url is not None and str(url).strip():
+ resolved = validate_directory_fetch_url(url)
+ payload = fetch_directory_payload(resolved, timeout=timeout)
+ rows = rows_from_payload(payload)
+ return transform_directory_rows(rows), resolved
+
+ row_lists: list[list] = []
+ used: list[str] = []
+ errors: list[str] = []
+ for candidate in DEFAULT_DIRECTORY_URLS:
+ try:
+ payload = fetch_directory_payload(candidate, timeout=timeout)
+ row_lists.append(rows_from_payload(payload))
+ used.append(candidate)
+ except Exception as exc:
+ errors.append(f"{candidate}: {exc}")
+ if not used:
+ msg = "; ".join(errors) if errors else "No directory URLs configured"
+ raise ValueError(msg)
+ rows = _merge_directory_rows(row_lists)
+ return transform_directory_rows(rows), " + ".join(used)
+
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
diff --git a/tests/backend/http_api_response_registry.py b/tests/backend/http_api_response_registry.py
index 3061ecdd..a83b0028 100644
--- a/tests/backend/http_api_response_registry.py
+++ b/tests/backend/http_api_response_registry.py
@@ -658,9 +658,6 @@ HTTP_JSON_GET_CONTRACT_EXCLUDED: tuple[str, ...] = (
"/api/v1/stickers/export",
"/api/v1/sticker-packs/{pack_id}/export",
"/api/v1/telephone/contacts/export",
- "/api/v1/tools/rnode/download_firmware",
- "/api/v1/tools/rnode/latest_release",
- "/api/v1/tools/micron-parser-go-release",
"/api/v1/filesync/content",
"/api/v1/favourites/layout",
"/api/v1/map/overlays",
diff --git a/tests/backend/test_announce_manager_extended.py b/tests/backend/test_announce_manager_extended.py
index 4d5d53ef..0575ce76 100644
--- a/tests/backend/test_announce_manager_extended.py
+++ b/tests/backend/test_announce_manager_extended.py
@@ -44,6 +44,28 @@ def test_upsert_announce(mock_db):
assert data["app_data"] == base64.b64encode(b"app_data").decode("utf-8")
+def test_upsert_omits_oversized_app_data(mock_db):
+ from meshchatx.src.backend.announce_manager import MAX_ANNOUNCE_APP_DATA_BYTES
+
+ manager = AnnounceManager(mock_db)
+ identity = MagicMock()
+ identity.hash.hex.return_value = "id_hash"
+ identity.get_public_key.return_value = b"pub_key"
+ manager.upsert_announce(
+ None,
+ identity,
+ b"dest_hash",
+ "aspect",
+ b"z" * (MAX_ANNOUNCE_APP_DATA_BYTES + 1),
+ None,
+ )
+ mock_db.announces.upsert_announce.assert_called_once()
+ data = mock_db.announces.upsert_announce.call_args[0][0]
+ assert "app_data" not in data
+ assert data["destination_hash"] == b"dest_hash".hex()
+ assert data["aspect"] == "aspect"
+
+
def test_upsert_skips_when_store_disabled_for_aspect(mock_db):
config = MagicMock()
config.announce_store_lxmf_delivery = MagicMock()
diff --git a/tests/backend/test_app_security_features.py b/tests/backend/test_app_security_features.py
index 16f555ac..c63b3995 100644
--- a/tests/backend/test_app_security_features.py
+++ b/tests/backend/test_app_security_features.py
@@ -1,7 +1,6 @@
# SPDX-License-Identifier: 0BSD
import secrets
-from unittest.mock import MagicMock
import bcrypt
import pytest
@@ -154,20 +153,3 @@ async def test_privacy_mode_blocks_map_export(mock_app):
headers=headers,
)
assert r.status == 403
-
-
-@pytest.mark.asyncio
-@pytest.mark.usefixtures("require_loopback_tcp")
-async def test_privacy_mode_blocks_repository_refresh(mock_app):
- mock_app.config.privacy_mode_enabled.set(True)
- mock_app.repository_server_manager = MagicMock()
- aio_app = _make_aio_app(mock_app, use_https=False)
-
- async with TestClient(TestServer(aio_app)) as client:
- headers = await fetch_api_csrf_headers(client)
- r = await client.post(
- "/api/v1/repository-server/refresh-bundled",
- headers=headers,
- )
- assert r.status == 403
- mock_app.repository_server_manager.refresh_bundled_wheels.assert_not_called()
diff --git a/tests/backend/test_community_interfaces.py b/tests/backend/test_community_interfaces.py
index 81300d9c..98d9d68c 100644
--- a/tests/backend/test_community_interfaces.py
+++ b/tests/backend/test_community_interfaces.py
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: 0BSD
import json
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
import pytest
@@ -22,7 +22,6 @@ async def test_community_interfaces_manager_no_probe():
@pytest.mark.asyncio
async def test_rnstatus_integration_simulated():
- # Simulate how rnstatus would see these interfaces if they were added
mock_reticulum = MagicMock()
mock_reticulum.get_interface_stats.return_value = {
"interfaces": [
@@ -61,31 +60,7 @@ async def test_community_interfaces_static_list():
@pytest.mark.asyncio
-async def test_community_interfaces_cache_used_when_no_public_override(tmp_path):
- cache = tmp_path / "community_interfaces_cache.json"
- cache.write_text(
- json.dumps(
- {
- "interfaces": [
- {
- "name": "CacheOnly",
- "type": "TCPClientInterface",
- "target_host": "10.0.0.1",
- "target_port": 4242,
- },
- ],
- },
- ),
- encoding="utf-8",
- )
- manager = CommunityInterfacesManager(public_override_path=None, cache_path=cache)
- ifaces = await manager.get_interfaces()
- assert len(ifaces) == 1
- assert ifaces[0]["name"] == "CacheOnly"
-
-
-@pytest.mark.asyncio
-async def test_community_interfaces_public_override_beats_cache(tmp_path):
+async def test_community_interfaces_public_override_beats_bundled(tmp_path):
public = tmp_path / "public.json"
public.write_text(
json.dumps(
@@ -102,50 +77,7 @@ async def test_community_interfaces_public_override_beats_cache(tmp_path):
),
encoding="utf-8",
)
- cache = tmp_path / "cache.json"
- cache.write_text(
- json.dumps(
- {
- "interfaces": [
- {
- "name": "FromCache",
- "type": "TCPClientInterface",
- "target_host": "10.0.0.3",
- "target_port": 4242,
- },
- ],
- },
- ),
- encoding="utf-8",
- )
- manager = CommunityInterfacesManager(
- public_override_path=str(public),
- cache_path=str(cache),
- )
+ manager = CommunityInterfacesManager(public_override_path=str(public))
ifaces = await manager.get_interfaces()
assert len(ifaces) == 1
assert ifaces[0]["name"] == "FromPublic"
-
-
-def test_refresh_from_directory_writes_cache(tmp_path):
- fake = [
- {
- "name": "FromNet",
- "type": "TCPClientInterface",
- "target_host": "9.9.9.9",
- "target_port": 4242,
- },
- ]
- cache = tmp_path / "community_interfaces_cache.json"
- manager = CommunityInterfacesManager(public_override_path=None, cache_path=cache)
- with patch(
- "meshchatx.src.backend.community_interfaces_directory.build_interfaces_from_directory_url",
- return_value=(fake, "https://example.test/list"),
- ) as mock_build:
- out = manager.refresh_from_directory()
- mock_build.assert_called_once()
- assert out["count"] == 1
- assert out["source"] == "https://example.test/list"
- assert cache.is_file()
- manager2 = CommunityInterfacesManager(public_override_path=None, cache_path=cache)
- assert manager2.interfaces[0]["name"] == "FromNet"
diff --git a/tests/backend/test_community_interfaces_directory.py b/tests/backend/test_community_interfaces_directory.py
index dfa2435e..310848d7 100644
--- a/tests/backend/test_community_interfaces_directory.py
+++ b/tests/backend/test_community_interfaces_directory.py
@@ -1,144 +1,13 @@
# SPDX-License-Identifier: 0BSD
-import threading
-import urllib.error
-import urllib.request
-from http.server import BaseHTTPRequestHandler, HTTPServer
-
import pytest
from hypothesis import given, settings
from hypothesis import strategies as st
from meshchatx.src.backend.community_interfaces_directory import (
- DEFAULT_DIRECTORY_URLS,
- DEFAULT_DISCOVERED_URL,
- DEFAULT_SUBMITTED_URL,
- fetch_directory_payload,
rows_from_payload,
transform_directory_rows,
- validate_directory_fetch_url,
-)
-
-
-def test_default_url_is_submitted_online():
- assert "submitted" in DEFAULT_SUBMITTED_URL
- assert "status=online" in DEFAULT_SUBMITTED_URL
- assert "discovered" in DEFAULT_DISCOVERED_URL
- assert DEFAULT_SUBMITTED_URL in DEFAULT_DIRECTORY_URLS
- assert DEFAULT_DISCOVERED_URL in DEFAULT_DIRECTORY_URLS
- assert "search=" not in DEFAULT_SUBMITTED_URL
- assert "type=" not in DEFAULT_SUBMITTED_URL
-
-
-def test_validate_directory_fetch_url_accepts_default_host():
- u = "https://directory.rns.recipes/api/foo?bar=1"
- assert validate_directory_fetch_url(u) == u
-
-
-@pytest.mark.parametrize(
- "bad",
- [
- "http://directory.rns.recipes/api",
- "https://127.0.0.1/",
- "https://metadata.internal/",
- "ftp://directory.rns.recipes/",
- "https://evil.com/https://directory.rns.recipes/",
- "https://user:pass@directory.rns.recipes/",
- "https://directory.rns.recipes@evil.com/",
- "https://not-directory.rns.recipes.example/api",
- ],
)
-def test_validate_directory_fetch_url_rejects_ssrf(bad):
- with pytest.raises(ValueError):
- validate_directory_fetch_url(bad)
-
-
-class _Redirect302Handler(BaseHTTPRequestHandler):
- def do_GET(self):
- self.send_response(302)
- self.send_header("Location", "http://127.0.0.1:9/")
- self.end_headers()
-
- def log_message(self, *args):
- return
-
-
-class _Json200Handler(BaseHTTPRequestHandler):
- def do_GET(self):
- body = b'{"data":[]}'
- self.send_response(200)
- self.send_header("Content-Type", "application/json")
- self.send_header("Content-Length", str(len(body)))
- self.end_headers()
- self.wfile.write(body)
-
- def log_message(self, *args):
- return
-
-
-@pytest.fixture
-def redirect_http_port():
- srv = HTTPServer(("127.0.0.1", 0), _Redirect302Handler)
- thread = threading.Thread(target=srv.serve_forever, daemon=True)
- thread.start()
- port = srv.server_address[1]
- yield port
- srv.shutdown()
-
-
-@pytest.fixture
-def json_http_port():
- srv = HTTPServer(("127.0.0.1", 0), _Json200Handler)
- thread = threading.Thread(target=srv.serve_forever, daemon=True)
- thread.start()
- port = srv.server_address[1]
- yield port
- srv.shutdown()
-
-
-def test_directory_fetch_opener_blocks_http_redirect(redirect_http_port):
- from meshchatx.src.backend.community_interfaces_directory import (
- _DIRECTORY_FETCH_OPENER,
- )
-
- req = urllib.request.Request(f"http://127.0.0.1:{redirect_http_port}/")
- with pytest.raises(urllib.error.HTTPError) as ei:
- _DIRECTORY_FETCH_OPENER.open(req, timeout=3)
- assert ei.value.code == 302
-
-
-def test_fetch_directory_payload_reads_json_when_no_redirect(
- monkeypatch,
- json_http_port,
-):
- import meshchatx.src.backend.community_interfaces_directory as cid
-
- monkeypatch.setattr(
- cid,
- "validate_directory_fetch_url",
- lambda url: url,
- )
- out = fetch_directory_payload(
- f"http://127.0.0.1:{json_http_port}/x",
- timeout=3,
- )
- assert out == {"data": []}
-
-
-def test_fetch_directory_payload_raises_on_redirect(monkeypatch, redirect_http_port):
- import meshchatx.src.backend.community_interfaces_directory as cid
-
- monkeypatch.setattr(
- cid,
- "validate_directory_fetch_url",
- lambda url: url,
- )
- with pytest.raises(urllib.error.HTTPError) as ei:
- fetch_directory_payload(
- f"http://127.0.0.1:{redirect_http_port}/",
- timeout=3,
- )
- assert ei.value.code == 302
def test_rows_from_payload_dict_data():
diff --git a/tests/backend/test_map_geo_validator.py b/tests/backend/test_map_geo_validator.py
index e9f070d3..a7b09a56 100644
--- a/tests/backend/test_map_geo_validator.py
+++ b/tests/backend/test_map_geo_validator.py
@@ -112,6 +112,22 @@ def test_validate_kmz_ok():
assert result.format == "kmz"
+def test_validate_kmz_too_large():
+ kml = b"""<?xml version="1.0"?>
+ <kml xmlns="http://www.opengis.net/kml/2.2">
+ <Document><Placemark><Point><coordinates>1,2,0</coordinates></Point></Placemark></Document>
+ </kml>"""
+ data = _kmz_with_kml(kml)
+ with pytest.raises(GeoValidationError) as exc:
+ validate_geo_bytes(
+ data,
+ max_bytes=8,
+ max_features=100,
+ max_kmz_uncompressed_bytes=1024 * 1024,
+ )
+ assert exc.value.code == "file_too_large"
+
+
def test_validate_kmz_missing_kml():
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
diff --git a/tests/backend/test_nomadnet_downloader.py b/tests/backend/test_nomadnet_downloader.py
index a9ca6e89..3a5d379f 100644
--- a/tests/backend/test_nomadnet_downloader.py
+++ b/tests/backend/test_nomadnet_downloader.py
@@ -6,6 +6,7 @@ import pytest
import RNS
from meshchatx.src.backend.nomadnet_downloader import (
+ MAX_NOMAD_PAGE_BYTES,
NomadnetDownloader,
NomadnetFileDownloader,
NomadnetPageDownloader,
@@ -139,6 +140,43 @@ def test_page_downloader_empty_response():
on_ok.assert_not_called()
+def test_page_downloader_rejects_oversized_body():
+ on_ok = MagicMock()
+ on_fail = MagicMock()
+ pd = NomadnetPageDownloader(
+ b"ab" * 8,
+ "/page.mu",
+ None,
+ on_ok,
+ on_fail,
+ MagicMock(),
+ )
+ rr = MagicMock()
+ rr.response = b"x" * (MAX_NOMAD_PAGE_BYTES + 1)
+ pd.on_download_success(rr)
+ on_fail.assert_called_once_with("page_too_large")
+ on_ok.assert_not_called()
+
+
+def test_page_downloader_accepts_body_at_cap():
+ on_ok = MagicMock()
+ on_fail = MagicMock()
+ pd = NomadnetPageDownloader(
+ b"ab" * 8,
+ "/page.mu",
+ None,
+ on_ok,
+ on_fail,
+ MagicMock(),
+ )
+ rr = MagicMock()
+ rr.response = b"y" * MAX_NOMAD_PAGE_BYTES
+ pd.on_download_success(rr)
+ on_ok.assert_called_once()
+ on_fail.assert_not_called()
+ assert on_ok.call_args[0][0] == "y" * MAX_NOMAD_PAGE_BYTES
+
+
def test_file_downloader_list_response_short_list_no_crash():
on_ok = MagicMock()
on_fail = MagicMock()
diff --git a/tests/backend/test_outbound_http_allowlist.py b/tests/backend/test_outbound_http_allowlist.py
new file mode 100644
index 00000000..765e3871
--- /dev/null
+++ b/tests/backend/test_outbound_http_allowlist.py
@@ -0,0 +1,74 @@
+# SPDX-License-Identifier: 0BSD
+
+"""New backend clearnet fetches must be listed here and gated by privacy mode.
+
+Do not add a file to KNOWN_CLEARNET_FETCH_FILES until it calls
+ensure_outbound_http_allowed, http_url_guard, or _require_outbound_http
+(or is the RNS HTTPInterface, which is mesh transport, not app clearnet).
+"""
+
+from pathlib import Path
+
+BACKEND_ROOT = Path("meshchatx/src/backend")
+
+FETCH_MARKERS = (
+ "aiohttp.ClientSession",
+ "httpx.Client(",
+ "httpx.AsyncClient(",
+ "urllib.request.urlopen",
+ "urllib.request.Request(",
+)
+
+KNOWN_CLEARNET_FETCH_FILES = frozenset(
+ {
+ "data/interfaces/HTTPInterface.py",
+ "map_manager.py",
+ "repository_server_manager.py",
+ "translator_handler.py",
+ },
+)
+
+MESH_TRANSPORT_FETCH_FILES = frozenset(
+ {
+ "data/interfaces/HTTPInterface.py",
+ },
+)
+
+PRIVACY_GATE_MARKERS = (
+ "ensure_outbound_http_allowed",
+ "http_url_guard",
+ "_require_outbound_http",
+ "OutboundHttpBlockedError",
+)
+
+
+def _rel(path: Path) -> str:
+ return str(path.relative_to(BACKEND_ROOT)).replace("\\", "/")
+
+
+def test_new_backend_clearnet_fetches_are_allowlisted():
+ found: set[str] = set()
+ for path in BACKEND_ROOT.rglob("*.py"):
+ text = path.read_text(encoding="utf-8")
+ if any(marker in text for marker in FETCH_MARKERS):
+ found.add(_rel(path))
+ assert found == set(KNOWN_CLEARNET_FETCH_FILES), (
+ "New backend httpx/urllib/aiohttp client usage must call "
+ "ensure_outbound_http_allowed or http_url_guard, then be added to "
+ "KNOWN_CLEARNET_FETCH_FILES. Extra: "
+ f"{sorted(found - KNOWN_CLEARNET_FETCH_FILES)}. Missing from disk: "
+ f"{sorted(KNOWN_CLEARNET_FETCH_FILES - found)}"
+ )
+
+
+def test_allowlisted_app_fetches_mention_a_privacy_gate():
+ for rel in sorted(KNOWN_CLEARNET_FETCH_FILES - MESH_TRANSPORT_FETCH_FILES):
+ text = (BACKEND_ROOT / rel).read_text(encoding="utf-8")
+ if any(marker in text for marker in PRIVACY_GATE_MARKERS):
+ continue
+ # Route-gated helpers: the HTTP route must still call the gate.
+ assert rel in {
+ "map_manager.py",
+ "repository_server_manager.py",
+ "translator_handler.py",
+ }, f"{rel} opens a clearnet socket without a privacy-mode or URL guard"
diff --git a/tests/backend/test_repository_server_manager.py b/tests/backend/test_repository_server_manager.py
index 03fead60..be0a1f26 100644
--- a/tests/backend/test_repository_server_manager.py
+++ b/tests/backend/test_repository_server_manager.py
@@ -2,7 +2,6 @@
import time
import urllib.request
-from pathlib import Path
from unittest.mock import patch
import pytest
@@ -153,18 +152,6 @@ def test_save_rejects_invalid_upload_filenames(tmp_path, name):
assert not ok
-@patch(
- "meshchatx.src.backend.repository_server_manager.download_bundled_wheels_to_directory",
-)
-def test_refresh_invokes_bundled_downloader(mock_dl, tmp_path):
- mock_dl.return_value = {"ok": True, "downloaded": ["rns"], "failed": {}}
- mgr = RepositoryServerManager(str(tmp_path))
- out = mgr.refresh_bundled_wheels()
- assert out["ok"] is True
- mock_dl.assert_called_once()
- assert mock_dl.call_args.kwargs.get("on_package") is not None
-
-
@patch(
"meshchatx.src.backend.repository_server_manager.stage_local_meshchatx_wheel_into_bundled_dir",
return_value=None,
@@ -209,58 +196,6 @@ def test_download_bundled_wheels_records_pypi_failures(
assert mock_pypi.call_count == n
-@patch(
- "meshchatx.src.backend.repository_server_manager.stage_local_meshchatx_wheel_into_bundled_dir",
- return_value=None,
-)
-@patch("meshchatx.src.backend.repository_server_manager._download_wheel_via_pypi_index")
-def test_refresh_bundled_wheels_fails_when_pypi_unavailable(
- mock_pypi,
- _mock_stage,
- tmp_path,
- monkeypatch,
-):
- monkeypatch.setenv("MESHCHAT_REPOSITORY_EXTRA_PIP", "")
- mock_pypi.return_value = (False, "offline")
- mgr = RepositoryServerManager(str(tmp_path))
- out = mgr.refresh_bundled_wheels()
- assert out["ok"] is False
- assert not out["downloaded"]
-
-
-@patch(
- "meshchatx.src.backend.repository_server_manager.stage_local_meshchatx_wheel_into_bundled_dir",
- return_value=None,
-)
-@patch("meshchatx.src.backend.repository_server_manager._download_wheel_via_pypi_index")
-def test_refresh_preserves_existing_wheels_when_pypi_fails(
- mock_pypi,
- _mock_stage,
- tmp_path,
- monkeypatch,
-):
- monkeypatch.setenv("MESHCHAT_REPOSITORY_EXTRA_PIP", "")
- mock_pypi.return_value = (False, "offline")
- mgr = RepositoryServerManager(str(tmp_path))
- keep = Path(mgr.bundled_dir) / "keep-me.whl"
- keep.write_bytes(b"wheel")
- out = mgr.refresh_bundled_wheels()
- assert out["ok"] is False
- assert keep.exists()
- assert keep.read_bytes() == b"wheel"
-
-
-def test_refresh_rejects_concurrent_calls(tmp_path):
- mgr = RepositoryServerManager(str(tmp_path))
- assert mgr._refresh_lock.acquire(blocking=False)
- try:
- out = mgr.refresh_bundled_wheels()
- assert out["ok"] is False
- assert out.get("error") == "refresh_already_running"
- finally:
- mgr._refresh_lock.release()
-
-
def test_http_start_stop_and_status(tmp_path):
mgr = RepositoryServerManager(str(tmp_path))
assert mgr.status()["http"]["running"] is False
diff --git a/tests/backend/test_rnode_download_firmware.py b/tests/backend/test_rnode_download_firmware.py
deleted file mode 100644
index 218dfc24..00000000
--- a/tests/backend/test_rnode_download_firmware.py
+++ /dev/null
@@ -1,418 +0,0 @@
-# SPDX-License-Identifier: 0BSD
-
-"""HTTP integration tests for the RNode firmware proxy endpoint."""
-
-from __future__ import annotations
-
-from contextlib import asynccontextmanager
-from unittest.mock import MagicMock, patch
-
-import pytest
-from aiohttp import web
-from aiohttp.test_utils import TestClient, TestServer
-
-pytestmark = pytest.mark.usefixtures("require_loopback_tcp")
-
-
-def _build_aio_app(app):
- routes = web.RouteTableDef()
- auth_mw, mime_mw, sec_mw, csrf_mw, ip_mw, demo_mw = app._define_routes(routes)
- aio_app = web.Application(
- middlewares=[auth_mw, mime_mw, sec_mw, csrf_mw, ip_mw, demo_mw]
- )
- aio_app.add_routes(routes)
- return aio_app
-
-
-@pytest.fixture
-def web_app(mock_app):
- mock_app.current_context.running = True
- mock_app.config.auth_enabled.set(False)
- return mock_app
-
-
-class _FakeResponse:
- def __init__(self, status: int, body: bytes, url: str = "https://github.com/x"):
- self.status = status
- self._body = body
- self.url = url
-
- async def read(self):
- return self._body
-
-
-class _FakeSession:
- def __init__(self, status: int, body: bytes, final_url: str | None = None):
- self._status = status
- self._body = body
- self._final_url = final_url
- self.requested_urls: list[str] = []
-
- async def __aenter__(self):
- return self
-
- async def __aexit__(self, exc_type, exc, tb):
- return False
-
- def get(self, url, allow_redirects=True, headers=None):
- self.requested_urls.append(url)
- status = self._status
- body = self._body
- final_url = self._final_url or url
-
- @asynccontextmanager
- async def _cm():
- yield _FakeResponse(status, body, url=final_url)
-
- return _cm()
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_requires_url(web_app):
- aio_app = _build_aio_app(web_app)
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get("/api/v1/tools/rnode/download_firmware")
- assert r.status == 400
- body = await r.json()
- assert "URL" in body["error"]
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_rejects_disallowed_redirect_target(web_app):
- aio_app = _build_aio_app(web_app)
- fake_session = _FakeSession(
- 200,
- b"PK\x03\x04ssrf",
- final_url="http://127.0.0.1:9337/secret",
- )
-
- with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip",
- },
- )
- assert r.status == 403
- body = await r.json()
- assert "redirect" in body["error"].lower()
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_allows_codeload_redirect(web_app):
- aio_app = _build_aio_app(web_app)
- fake_zip = b"PK\x03\x04ok"
- final = "https://codeload.github.com/owner/repo/zip/refs/tags/v1"
- fake_session = _FakeSession(200, fake_zip, final_url=final)
-
- with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/owner/repo/archive/refs/tags/v1.zip",
- },
- )
- assert r.status == 200
- assert await r.read() == fake_zip
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_returns_zip_for_allowed_url(web_app):
- aio_app = _build_aio_app(web_app)
- fake_zip = b"PK\x03\x04fake-zip-bytes"
- fake_session = _FakeSession(200, fake_zip)
-
- with patch(
- "aiohttp.ClientSession",
- MagicMock(return_value=fake_session),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip",
- },
- )
- assert r.status == 200
- assert r.headers.get("Content-Type", "").startswith("application/zip")
- assert r.headers.get("Content-Disposition", "").endswith('"firmware.zip"')
- data = await r.read()
- assert data == fake_zip
- assert fake_session.requested_urls == [
- "https://github.com/owner/repo/releases/download/v1/firmware.zip",
- ]
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_strips_crlf_from_content_disposition(web_app):
- aio_app = _build_aio_app(web_app)
- fake_zip = b"PK\x03\x04hdr"
- fake_session = _FakeSession(200, fake_zip)
- hostile_name = 'evil.zip"\r\nX-Injected: yes'
- url = "https://github.com/owner/repo/releases/download/v1/" + hostile_name
-
- with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={"url": url},
- )
- assert r.status == 200
- disp = r.headers.get("Content-Disposition", "")
- prefix = 'attachment; filename="'
- assert disp.startswith(prefix)
- assert disp.endswith('"')
- inner = disp[len(prefix) : -1]
- assert "\r" not in disp
- assert "\n" not in disp
- assert '"' not in inner
- assert await r.read() == fake_zip
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_propagates_upstream_error_status(web_app):
- aio_app = _build_aio_app(web_app)
- fake_session = _FakeSession(404, b"")
-
- with patch(
- "aiohttp.ClientSession",
- MagicMock(return_value=fake_session),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/markqvist/RNode_Firmware/releases/download/v1/firmware.zip",
- },
- )
- assert r.status == 404
- body = await r.json()
- assert "Failed to download" in body["error"]
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_returns_500_on_exception(web_app):
- aio_app = _build_aio_app(web_app)
-
- with patch(
- "aiohttp.ClientSession",
- MagicMock(side_effect=RuntimeError("network down")),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={
- "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip",
- },
- )
- assert r.status == 500
- body = await r.json()
- assert "network down" in body["error"]
-
-
-class _FakeJsonResponse:
- def __init__(self, status: int, payload):
- self.status = status
- self._payload = payload
-
- async def json(self, content_type=None):
- return self._payload
-
- async def read(self):
- import json
-
- return json.dumps(self._payload).encode("utf-8")
-
-
-class _FakeJsonSession:
- def __init__(self, status: int, payload):
- self._status = status
- self._payload = payload
- self.requested_urls: list[str] = []
- self.last_headers = None
-
- async def __aenter__(self):
- return self
-
- async def __aexit__(self, exc_type, exc, tb):
- return False
-
- def get(self, url, allow_redirects=True, headers=None):
- self.requested_urls.append(url)
- self.last_headers = headers
- status = self._status
- payload = self._payload
-
- @asynccontextmanager
- async def _cm():
- yield _FakeJsonResponse(status, payload)
-
- return _cm()
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_accepts_objects_githubusercontent_url(web_app):
- aio_app = _build_aio_app(web_app)
- fake_zip = b"PK\x03\x04x"
- fake_session = _FakeSession(200, fake_zip)
- asset_url = (
- "https://objects.githubusercontent.com/github-production-release-asset/1/2/3"
- "?response-content-disposition=attachment%3B%20filename%3Dfw.zip"
- )
-
- with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={"url": asset_url},
- )
- assert r.status == 200
- assert fake_session.requested_urls == [asset_url]
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_accepts_release_assets_githubusercontent_url(web_app):
- aio_app = _build_aio_app(web_app)
- fake_zip = b"PK\x03\x04y"
- fake_session = _FakeSession(200, fake_zip)
- asset_url = "https://release-assets.githubusercontent.com/github-production-release-asset/9/8/7/fw.zip"
-
- with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={"url": asset_url},
- )
- assert r.status == 200
- data = await r.read()
- assert data == fake_zip
-
-
-@pytest.mark.asyncio
-async def test_download_firmware_accepts_configured_gitea_base_url(web_app):
- aio_app = _build_aio_app(web_app)
- web_app.config.gitea_base_url.set("https://gitea.custom.example")
- fake_zip = b"PK\x03\x04z"
- fake_session = _FakeSession(200, fake_zip)
- asset_url = (
- "https://gitea.custom.example/someorg/somerepo/releases/download/v1/x.zip"
- )
-
- with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/download_firmware",
- params={"url": asset_url},
- )
- assert r.status == 200
- assert fake_session.requested_urls == [asset_url]
-
-
-@pytest.mark.asyncio
-async def test_latest_release_returns_proxied_payload(web_app):
- aio_app = _build_aio_app(web_app)
- payload = {
- "tag_name": "v1.83",
- "assets": [
- {
- "name": "rnode_firmware_heltec32v3.zip",
- "browser_download_url": "https://x/rnode.zip",
- },
- ],
- }
- fake_session = _FakeJsonSession(200, payload)
-
- with patch(
- "aiohttp.ClientSession",
- MagicMock(return_value=fake_session),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get("/api/v1/tools/rnode/latest_release")
- assert r.status == 200
- body = await r.json()
- assert body == payload
- assert fake_session.requested_urls[0] == (
- "https://api.github.com/repos/markqvist/RNode_Firmware/releases/latest"
- )
- assert fake_session.last_headers is not None
- assert (
- fake_session.last_headers.get("Accept") == "application/vnd.github+json"
- )
- assert fake_session.last_headers.get("X-GitHub-Api-Version") == "2022-11-28"
- assert "MeshChatX-RNodeFlasher" in fake_session.last_headers.get(
- "User-Agent",
- "",
- )
-
-
-@pytest.mark.asyncio
-async def test_latest_release_uses_repo_query_param(web_app):
- aio_app = _build_aio_app(web_app)
- fake_session = _FakeJsonSession(200, {"tag_name": "v0"})
-
- with patch(
- "aiohttp.ClientSession",
- MagicMock(return_value=fake_session),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get(
- "/api/v1/tools/rnode/latest_release",
- params={"repo": "Some/Other_Repo"},
- )
- assert r.status == 200
- assert fake_session.requested_urls[0] == (
- "https://api.github.com/repos/Some/Other_Repo/releases/latest"
- )
-
-
-@pytest.mark.asyncio
-async def test_latest_release_rejects_invalid_repo(web_app):
- aio_app = _build_aio_app(web_app)
- async with TestClient(TestServer(aio_app)) as client:
- for repo in (
- "no-slash",
- "../etc/passwd",
- "evil repo/x",
- "bad?repo/x",
- "too/many/slashes",
- "@bad/name",
- "/leading/slash",
- "trailing/",
- ):
- r = await client.get(
- "/api/v1/tools/rnode/latest_release",
- params={"repo": repo},
- )
- assert r.status == 400, f"expected 400 for repo={repo!r}"
-
-
-@pytest.mark.asyncio
-async def test_latest_release_propagates_upstream_status(web_app):
- aio_app = _build_aio_app(web_app)
- fake_session = _FakeJsonSession(404, {})
-
- with patch(
- "aiohttp.ClientSession",
- MagicMock(return_value=fake_session),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get("/api/v1/tools/rnode/latest_release")
- assert r.status == 404
- body = await r.json()
- assert "Failed to fetch release" in body["error"]
-
-
-@pytest.mark.asyncio
-async def test_latest_release_returns_500_on_exception(web_app):
- aio_app = _build_aio_app(web_app)
- with patch(
- "aiohttp.ClientSession",
- MagicMock(side_effect=RuntimeError("dns down")),
- ):
- async with TestClient(TestServer(aio_app)) as client:
- r = await client.get("/api/v1/tools/rnode/latest_release")
- assert r.status == 500
- body = await r.json()
- assert "dns down" in body["error"]
diff --git a/tests/backend/test_rnode_download_url_oracle.py b/tests/backend/test_rnode_download_url_oracle.py
deleted file mode 100644
index d35890b4..00000000
--- a/tests/backend/test_rnode_download_url_oracle.py
+++ /dev/null
@@ -1,32 +0,0 @@
-# SPDX-License-Identifier: 0BSD
-
-"""Oracle: RNode firmware download URL allowlist rejects lookalike hosts."""
-
-from __future__ import annotations
-
-
-def _url_allowed(url: str, allowed_prefixes: list[str]) -> bool:
- return any(url.startswith(a) for a in allowed_prefixes)
-
-
-def test_rnode_download_prefix_oracle_rejects_lookalike_hosts():
- allowed = [
- "https://github.com/",
- "https://codeload.github.com/",
- "https://objects.githubusercontent.com/",
- "https://release-assets.githubusercontent.com/",
- ]
- # Accept
- assert _url_allowed(
- "https://github.com/markqvist/RNode_Firmware/releases/download/v1/x.zip",
- allowed,
- )
- assert _url_allowed(
- "https://objects.githubusercontent.com/github-production-release-asset-2e65be/1/x",
- allowed,
- )
- # Reject lookalikes / SSRF bait
- assert not _url_allowed("https://github.com.evil.example/markqvist/x.zip", allowed)
- assert not _url_allowed("https://evil.example/https://github.com/x.zip", allowed)
- assert not _url_allowed("http://127.0.0.1/firmware.zip", allowed)
- assert not _url_allowed("https://github.com.attacker/x", allowed)
diff --git a/tests/backend/test_rrc_security.py b/tests/backend/test_rrc_security.py
index be8422f3..9eaad9ea 100644
--- a/tests/backend/test_rrc_security.py
+++ b/tests/backend/test_rrc_security.py
@@ -233,6 +233,32 @@ def test_server_rejects_oversized_message_body():
assert "large" in out[0][1][proto.K_BODY].lower()
+def test_server_rejects_oversized_message_body_at_default_cap():
+ server = make_server()
+ assert server.max_msg_body_bytes == proto.DEFAULT_MAX_MSG_BYTES
+ link = FakeLink(FakeIdentity(b"\x01" * 16))
+ sess = add_session(server, link, link._identity.hash, nick="alice")
+ route(
+ server,
+ link,
+ sess,
+ proto.make_envelope(proto.T_JOIN, src=sess.peer, room="lobby"),
+ )
+ out = route(
+ server,
+ link,
+ sess,
+ proto.make_envelope(
+ proto.T_MSG,
+ src=sess.peer,
+ room="lobby",
+ body="x" * (proto.DEFAULT_MAX_MSG_BYTES + 1),
+ ),
+ )
+ assert out[0][1][proto.K_T] == proto.T_ERROR
+ assert "large" in out[0][1][proto.K_BODY].lower()
+
+
def test_server_rejects_message_to_unknown_room():
server = make_server()
link = FakeLink(FakeIdentity(b"\x02" * 16))
Served by rngit 1.5.2 - Generated in 0.06s